package com.xr.el.utils;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.AcknowledgedResponse;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregation;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.cat.aliases.AliasesRecord;
import co.elastic.clients.elasticsearch.cat.indices.IndicesRecord;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
import co.elastic.clients.elasticsearch.core.search.Highlight;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.indices.*;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import co.elastic.clients.util.ObjectBuilder;
import com.alibaba.fastjson.JSONObject;
import com.xr.el.annotation.DocId;
import com.xr.el.annotation.EsClass;
import com.xr.el.annotation.EsField;
import com.xr.el.entity.EsDataType;
import jakarta.json.JsonObject;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;
import java.lang.reflect.Field;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

@Slf4j
@Component
@Data
public class Es8Utils implements InitializingBean
{
    @Value("${elasticsearch.host}")
    private String host = "127.0.0.1";

    @Value("${elasticsearch.port}")
    private Integer port = 9200;

    private ElasticsearchClient client;

    private String indexPrefix;

    private String aliasPrefix;

    @Override
    public void afterPropertiesSet()
    {
        if (ObjectUtils.isEmpty(client))
        {
            RestClient build = RestClient.builder(new HttpHost(host, port)).build();
            ElasticsearchTransport transport =
                    new RestClientTransport(build, new JacksonJsonpMapper());
            // es 客户端
            this.client = new ElasticsearchClient(transport);
        }
        this.indexPrefix = "index-";
        this.aliasPrefix = "alias-";
    }

    /**
     * 根据注解的信息查询索引
     * @param tClass
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> boolean createIndex(Class<T> tClass) throws IOException {
        //  获得这个类里面的es注解，通过这个注解获取对应的索引信息和分片信息
        EsClass annotation = tClass.getAnnotation(EsClass.class);

        //  获取对应的索引名和别名
        String index = getClassIndexOrAlias(tClass,"index");
        String alias = getClassIndexOrAlias(tClass,"alias");

        List<String> indexes = getAllIndexes();
        List<String> aliases = getAllAliases();

        //  索引名或者别名以及存在
        if(indexes.contains(index) || aliases.contains(alias))
        {
            return false;
        }

        //  获得分片和副片
        int shards = annotation.shards();
        int replicas = annotation.replicas();

        StringBuilder stringBuilder = new StringBuilder("{");
        stringBuilder.append(
                "\"settings\": {\n"
                        + "    \"number_of_shards\": "
                        + shards
                        + ",\n"
                        + "    \"number_of_replicas\": "
                        + replicas
                        + "\n"
                        + "  },");
        stringBuilder.append("\"mappings\": {\n" + "    \"properties\": ");
        JSONObject fieldsJsonObject = new JSONObject();
        // "email": {
        //           "type": "keyword",
        //           "index": false
        //             ...
        //          },
        //  "info": {
        //          "type": "text",
        //          "analyzer": "standard"
        //             ...
        //          },
        for (Field field : tClass.getDeclaredFields())
        {
            field.setAccessible(true);
            JSONObject fieldJsonObject = new JSONObject();

            //  _version 需要通过 docId 来查询
            DocId DocId = field.getAnnotation(DocId.class);
            if (DocId != null)
            {
                fieldJsonObject.put("type", EsDataType.LONG.getType());
                fieldsJsonObject.put(field.getName(), fieldJsonObject);
                continue;
            }

            EsField fieldAnnotation = field.getAnnotation(EsField.class);
            if (!ObjectUtils.isEmpty(fieldAnnotation))
            {
                //  获得映射名字
                String name = ObjectUtils.isEmpty(fieldAnnotation.name()) ? field.getName() : fieldAnnotation.name();
                //  获得分词、搜索分词、数据类型,并填充到fieldJsonObject
                EsDataType type = fieldAnnotation.type();
                String analyzer = fieldAnnotation.analyzer();
                String searchAnalyzer = fieldAnnotation.searchAnalyzer();

                fieldJsonObject.put("type", type.getType());
                if (!"".equals(analyzer)) {
                    fieldJsonObject.put("analyzer", analyzer);
                }
                if (!"".equals(searchAnalyzer)) {
                    fieldJsonObject.put("search_analyzer", searchAnalyzer);
                }

                // 最后填充"email":{}
                fieldsJsonObject.put(name,fieldJsonObject);
            }

        }
        Assert.isTrue(fieldsJsonObject.size() > 0,"没有任何映射，请添加相关字段的注解");
        stringBuilder.append(fieldsJsonObject);
        stringBuilder.append("}}");

        Reader queryJson = new StringReader(stringBuilder.toString());

        CreateIndexRequest req =
                CreateIndexRequest.of(
                                b ->    b.index(index)
                                        .aliases(alias, Alias.of(a -> a.isWriteIndex(true)))
                                        .withJson(queryJson));

        return client.indices().create(req).acknowledged();
    }

    /**
     * 删除索引
     * @param indexName
     * @param <T>
     * @return
     * @throws IOException
     */
    public <T> boolean deleteIndex(String indexName) throws IOException
    {
        Assert.notNull(indexName,"索引名不能为空");
        DeleteIndexRequest deleteIndexRequest = DeleteIndexRequest.of(d -> d.index(indexName));
        DeleteIndexResponse delete = client.indices().delete(deleteIndexRequest);
        return delete.acknowledged();
    }


    //  查询索引的全部数据
    public <T> List<T> queryAll(Class<T> tClass) throws IOException
    {
        List<T> list = new ArrayList<>();
        String index = getClassIndexOrAlias(tClass,"index");
        SearchResponse<T> search = client.search(s -> s.index(index).query(q -> q.matchAll(m -> m)), tClass);
        for (Hit<T> hit : search.hits().hits()) {
            list.add(hit.source());
        }
        return list;
    }

    // 添加数据
    /**
     * @param o  数据源
     * @return
     */
    public <T> String addData(T o)
    {
        Object id = null;
        for (Field declaredField : o.getClass().getDeclaredFields())
        {
            declaredField.setAccessible(true);
            DocId annotation = declaredField.getAnnotation(DocId.class);
            if (annotation != null) {
                try {
                    id =  declaredField.get(o);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                }
            }
        }
        if (id == null) {
            id = UUID.randomUUID().toString();
        }
        IndexResponse response = null;

        try {
            IndexRequest.Builder<T> indexReqBuilder = new IndexRequest.Builder<>();
            indexReqBuilder.index(getClassIndexOrAlias(o.getClass(),"index"));
            indexReqBuilder.id(String.valueOf(id));
            indexReqBuilder.document(o);
            response = client.index(indexReqBuilder.build());
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert response != null;
        return response.id();
    }

    // 批量添加
    public <T> void addBatchData(List<T> list)
    {
        BulkRequest.Builder br = new BulkRequest.Builder();
        for (T o : list)
        {
            Object id = null;
            for (Field declaredField : o.getClass().getDeclaredFields())
            {
                declaredField.setAccessible(true);
                DocId annotation = declaredField.getAnnotation(DocId.class);
                if (annotation != null) {
                    try {
                        id =  declaredField.get(o);
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    }
                }
            }
            if (id == null) {
                id = UUID.randomUUID().getMostSignificantBits();
            }
            Object finalId = id;
            br.operations(
                    op ->
                            op.index(
                                    idx ->
                                            idx.index(getClassIndexOrAlias(o.getClass(),"index")).id(String.valueOf(finalId)).document(o)));
        }

        BulkResponse result = null;
        try {
            result = client.bulk(br.build());
        } catch (IOException e) {
            e.printStackTrace();
        }
        // Log errors, if any
        assert result != null;
        if (result.errors()) {
            log.error("Bulk had errors");
            for (BulkResponseItem item : result.items()) {
                if (item.error() != null) {
                    log.error(item.error().reason());
                }
            }
        }
    }

    public <T> T getDocInfoByDocId(Long docId, Class<T> tClass)
    {
        GetResponse<T> response = null;
        try {
            response = client.get(g -> g.index(getClassIndexOrAlias(tClass,"index")).id(String.valueOf(docId)), tClass);
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert response != null;
        return response.source();
    }

    //   Query query = Query.of(q -> q.match(m -> m.field("id").query(101L)));
    public <T> List<T> complexQuery(Query query, Class<T> tClass)
    {
        ArrayList<T> res = new ArrayList<>();

        Field[] declaredFields = tClass.getDeclaredFields();
        List<String> collectFields = Arrays.stream(declaredFields).map(Field::getName).collect(Collectors.toList());
        Highlight of = Highlight.of(h -> {
                    for (String field : collectFields) {
                        h.fields(
                                field
                                ,
                                h1 -> h1.preTags("<font color='red'>").postTags("</font>"));
                    }
                    return h;
                }
        );
        try {
            SearchResponse<T> resp = client.search(s -> s.index(getClassIndexOrAlias(tClass, "index"))
                                            .query(query).highlight(of), tClass);
            List<Hit<T>> hits = resp.hits().hits();
            for (Hit<T> hit : hits)
            {
                res.add(hit.source());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return res;
    }

    public  <T> void upDocId(Long docId,T o) throws Exception
    {
        //先查询出来
        Object docId1 = getDocInfoByDocId(docId, o.getClass());
        if (ObjectUtils.isEmpty(docId))
        {
            throw new Exception("没有docId:"+docId+"这条数据");
        }

        String index = getClassIndexOrAlias(o.getClass(),"index");

        client.update(UpdateRequest.of(d -> d.index(index).id(String.valueOf(docId)).doc(o)), o.getClass());
    }

    //  es8Utils.complexQueryAggregations(query1,q -> q.avg(m -> m.field("price")),userEsEntity.getClass());
    public <T> Aggregate complexQueryAggregations(Query query, Function<Aggregation.Builder, ObjectBuilder<Aggregation>> fn, Class<T> clazz)
    {

        SearchResponse<T> response = null;
        try {
            response = client.search(
                    s ->
                            s.index(getClassIndexOrAlias(clazz,"index"))
                                    .size(0) // 不需要显示数据 ,只想要聚合结果
                                    .query(query)
                                    .aggregations("aggregations", fn),
                                    clazz);
        } catch (IOException e) {
            e.printStackTrace();
        }

        assert response != null;
        return response.aggregations().get("aggregations");
    }

    // 查询全部索引
    public List<String> getAllIndexes()
    {
        List<IndicesRecord> indices = null;
        try {
            indices = client.cat().indices().valueBody();
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert indices != null;
        return indices.stream().map(IndicesRecord::index).collect(Collectors.toList());
    }

    //  查询全部别名
    private List<String> getAllAliases()
    {
        List<AliasesRecord> aliasesRecords = null;
        try {
            aliasesRecords = client.cat().aliases().valueBody();
        } catch (IOException e) {
            e.printStackTrace();
        }
        assert aliasesRecords != null;
        return aliasesRecords.stream().map(AliasesRecord::alias).collect(Collectors.toList());
    }

    /**
     * 获取索引名或者别名
     * @param tClass
     * @param type
     * @param <T>
     * @return
     */
    private <T> String getClassIndexOrAlias(Class<T> tClass, String type)
    {
        if (ObjectUtils.isEmpty(tClass))
        {
            return null;
        }
        EsClass annotation = tClass.getAnnotation(EsClass.class);
        if (type.equals("index"))
        {
            String index = annotation.index();
            index = StringUtils.isEmpty(index) ? tClass.getName().toLowerCase() : index.toLowerCase();
            return indexPrefix + index;
        }else
        {
            String alias = annotation.alias();
            alias = StringUtils.isEmpty(alias) ? tClass.getName().toLowerCase() : alias.toLowerCase();
            return aliasPrefix +alias;
        }
    }
}

